package org.jeecg.desen.run.service.impl;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.api.vo.Result;
import org.jeecg.desen.error.AbstractErrorService;
import org.jeecg.desen.run.component.DscConversionService;
import org.jeecg.desen.run.entity.DscSsc;
import org.jeecg.desen.run.entity.DscSscTableDetail;
import org.jeecg.desen.run.service.DesRunService;
import org.jeecg.desen.run.threadPool.DbHiveTaskExecutor;
import org.jeecg.modules.meta.entity.SysTableinfo;
import org.jeecg.modules.meta.service.ISysTableinfoService;
import org.jeecg.modules.meta.service.MetaSynService;
import org.jeecg.modules.run.entity.Run;
import org.jeecg.modules.run.entity.RunDetail;
import org.jeecg.modules.run.service.IRunDetailService;
import org.jeecg.modules.run.service.IRunService;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;

/**
 * 脱敏运行抽象类
 */
@Slf4j
public abstract class AbstractDesRunService extends AbstractErrorService implements DesRunService {

    protected Date runDate;
    protected Date endDate;


    @Autowired
    MetaSynService metaSynService;

    @Autowired
    IRunService runService;

    @Autowired
    IRunDetailService runDetailService;

    @Autowired
    ISysTableinfoService iSysTableinfoService;


    @Override
    public boolean reRun(String runId) {
        //运行监控状态验证
        if (false == verificationReRun()) {
            return false;
        }

        Run run = getRunEntity(runId);
        if (null == run || run.getFdStatus() != 2) {
            return false;
        }

        try {
            //根据运行id获取订阅id
            DscSsc dscSsc = getRunDscSsc(run.getFdSscid());
            //重跑
            reRunDscTable(dscSsc, run);
        } catch (Exception e) {
            log.info("运行错误-中止", e);
            return false;
        }
        //等待线程执行完毕 依赖任务处理
        log.info("运行结束");

        return true;
    }

    private void reRunDscTable(DscSsc dscSsc, Run run) {
        //设置参数正确的运行批次
        reRunDscTableThreadPool(dscSsc, run);
    }

    private void reRunDscTableThreadPool(DscSsc dscSsc, Run runEntity) {
        log.info("线程池执行开始:" + dscSsc.getSscid());
        runEntity.setFdStatus(1);//重新设置状态
        runEntity.setFdStartTime(new Date());//设置开始时间

        final CountDownLatch latch = new CountDownLatch(dscSsc.getTableDetailList().size());
        for (DscSscTableDetail dscSscTableDetail : dscSsc.getTableDetailList()) {
            if (null == dscSscTableDetail) {
                latch.countDown();
                continue;
            }
            //获取运行监控信息 如果已经成功则跳过
            RunDetail runDetail = getReRunDetail(dscSscTableDetail.getFdTableId(), runEntity.getFdBatchId());
            if (null == runDetail || runDetail.getFdStatus() != 2) {
                latch.countDown();
                continue;
            }
            runDetail.setFdStartTime(new Date());


            //失败的继续跑
            DbHiveTaskExecutor.execute("脱敏表数据操作:订阅id:" + dscSsc.getSscid() + ",表id=" + dscSscTableDetail.getFdTableId(), new Thread() {
                @Override
                public void run() {
                    //初始化详情表运行信息
                    try {
                        setWhere(dscSscTableDetail);//设置分区条件
                        setTableField(dscSscTableDetail);//设置表对应的字段信息
                        //执行单表操作
                        Result result = getComponent().conversionDB(dscSscTableDetail);
                        //处理单表结果
                        dealResult(result, runDetail, dscSscTableDetail);
                    } catch (Exception e) {
                        String error = "脱敏表-hive,执行错误。表信息=" + JSON.toJSONString(dscSscTableDetail);
                        try {
                            recordError(error, 2, e);
                            saveRunDetailEntity(getErrorRunDetail(runDetail, error + ",异常信息=" + subErrorString(e.getMessage())));
                        } catch (Exception e1) {
                            log.error("插入错误记录表错误!", e1);
                        }
                    }
                    //释放资源
                    dscSscTableDetail.setFieldDetailList(null);
                    //线程计数-1
                    latch.countDown();
                }
            });
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            log.info("脱敏线程池等待结束异常,订阅id" + dscSsc.getSscid(), 2);
        }

        savaEndRunEntity(runEntity, dscSsc);
        log.info("线程池执行结束:" + dscSsc.getSscid());
    }

    //根据运行id获取订阅运行信息
    abstract DscSsc getRunDscSsc(String sscId);


    private RunDetail getReRunDetail(String tableId, String runId) {
        QueryWrapper<RunDetail> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("fd_batch_id", runId);
        queryWrapper.eq("fd_tabid", tableId);
        return runDetailService.getOne(queryWrapper);
    }

    private Run getRunEntity(String runId) {
        return runService.getById(runId);
    }

    boolean verificationReRun() {
        if (metaSynService.isSyn()) {
            recordError("启动失败！isSyn同步正在进行,请通过参数查看当前进程！");
            return false;
        }
        return true;
    }


    //执行脱敏运算
    //把表数据脱敏插入到脱敏库
    @Override
    public boolean run() {
        //验证
        if (false == verification()) {
            return false;
        }

        try {
            //通过频次或者id获取任务对应订阅信息
            List<DscSsc> dscSsc = getDscSsc();
            //根据表为单位信息执行脱敏处理
            runDsc(dscSsc);
        } catch (Exception e) {
            log.info("运行错误-中止", e);
            return false;
        }
        //等待线程执行完毕 依赖任务处理
        log.info("运行结束");
        return true;
    }

    private void runDsc(List<DscSsc> dscSscList) {
        //如果有多个订阅信息 依次往脱敏队列里面塞
        for (DscSsc dscSsc : dscSscList) {
            runDscTable(dscSsc);
        }
    }

    private void runDscTable(DscSsc dscSsc) {
        Run runEntity = initRunEntity(dscSsc);
        saveRunEntity(runEntity);
        List<DscSscTableDetail> dscSscTableDetails = dscSsc.getTableDetailList();
        if (null == dscSscTableDetails || dscSscTableDetails.size() == 0) {
            String errorStr = "该订阅批次没有对应的表脱敏：订阅id=" + dscSsc.getSscid();
            log.info(errorStr);
            //插入run表
            saveRunEntity(getErrorRun(initRunEntity(dscSsc), errorStr));
            return;
        }

        //设置参数正确的运行批次
        runDscTableThreadPool(dscSsc, runEntity);
    }

    private void runDscTableThreadPool(DscSsc dscSsc, Run runEntity) {
        log.info("线程池执行开始:" + dscSsc.getSscid());

        final CountDownLatch latch = new CountDownLatch(dscSsc.getTableDetailList().size());
        for (DscSscTableDetail dscSscTableDetail : dscSsc.getTableDetailList()) {
            if (null == dscSscTableDetail) {
                latch.countDown();
                continue;
            }
            DbHiveTaskExecutor.execute("脱敏表数据操作:订阅id:" + dscSsc.getSscid() + ",表id=" + dscSscTableDetail.getFdTableId(), new Thread() {
                @Override
                public void run() {
                    //初始化详情表运行信息
                    RunDetail runDetail = initRunDetailEntity(dscSscTableDetail, runEntity);
                    try {
                        setWhere(dscSscTableDetail);//设置分区条件
                        setTableField(dscSscTableDetail);//设置表对应的字段信息
                        //执行单表操作
                        Result result = getComponent().conversionDB(dscSscTableDetail);
                        //处理单表结果
                        dealResult(result, runDetail, dscSscTableDetail);
                    } catch (Exception e) {
                        String error = "脱敏表-hive,执行错误。表信息=" + JSON.toJSONString(dscSscTableDetail);
                        try {
                            recordError(error, 2, e);
                            saveRunDetailEntity(getErrorRunDetail(runDetail, error + ",异常信息=" + subErrorString(e.getMessage())));
                        } catch (Exception e1) {
                            log.error("插入错误记录表错误!", e1);
                        }
                    }
                    //释放资源
                    dscSscTableDetail.setFieldDetailList(null);
                    //线程计数-1
                    latch.countDown();
                }
            });
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            log.info("脱敏线程池等待结束异常,订阅id" + dscSsc.getSscid(), 2);
        }

        savaEndRunEntity(runEntity, dscSsc);
        log.info("线程池执行结束:" + dscSsc.getSscid());
    }

    boolean verification() {
        if (metaSynService.isSyn()) {
            recordError("启动失败！isSyn同步正在进行,请通过参数查看当前进程！");
            return false;
        }
        return verificationParam();
    }

    private void dealResult(Result result, RunDetail runDetail, DscSscTableDetail dscSscTableDetail) {
        runDetail.setFdTabname(dscSscTableDetail.getFdTableName());
        if (false == result.isSuccess()) {
            dscSscTableDetail.setStatus(2);
            dscSscTableDetail.setError(result.getMessage());
            saveRunDetailEntity(getErrorRunDetail(runDetail, result.getMessage()));
        } else {
            saveRunDetailEntity(getEndRunDetail(runDetail));
        }
    }

    private RunDetail initRunDetailEntity(DscSscTableDetail dscSscTableDetail, Run runEntity) {
        RunDetail runDetail = new RunDetail();
        runDetail.setFdSscid(runEntity.getFdSscid());//订阅id
        runDetail.setFdBatchId(runEntity.getFdBatchId());//批次号
        runDetail.setFdStartTime(new Date());//开始时间
        runDetail.setFdStatus(3);//开始时间
        runDetail.setFdTabid(dscSscTableDetail.getFdTableId());//表格id
        runDetail.setFdTabname(dscSscTableDetail.getFdTableName());//表格名称
        runDetail.setCreateTime(new Date());//创建时间
        return runDetail;
    }

    private void savaEndRunEntity(Run runEntity, DscSsc dscSsc) {
        try {
            runEntity = setStatusRun(getEndRun(runEntity), dscSsc.getTableDetailList());
        } catch (Exception e) {
            runEntity = getErrorRun(runEntity, subErrorString(e.getMessage()));
        }
        saveRunEntity(runEntity);
    }

    private void saveRunEntity(Run run) {
        runService.saveOrUpdate(run);
    }

    private void saveRunDetailEntity(RunDetail runDetail) {
        runDetailService.saveOrUpdate(runDetail);
    }

    private RunDetail getErrorRunDetail(RunDetail runDetail, String errorStr) {
        runDetail = getEndRunDetail(runDetail);
        runDetail.setFdStatus(2);
        runDetail.setFdError(errorStr);
        return runDetail;
    }

    private Run getErrorRun(Run runEntity, String errorStr) {
        runEntity = getEndRun(runEntity);
        runEntity.setFdStatus(2);
        runEntity.setFdError(errorStr);
        return runEntity;
    }

    private Run setStatusRun(Run runEntity, List<DscSscTableDetail> dscSscTableDetails) {
        for (DscSscTableDetail dscSscTableDetail : dscSscTableDetails) {
            if (null != dscSscTableDetail.getStatus() && dscSscTableDetail.getStatus() == 2) {
                runEntity.setFdStatus(2);
                runEntity.setFdError(dscSscTableDetail.getError());
                return runEntity;
            }
        }
        return runEntity;
    }

    private Run getEndRun(Run runEntity) {
        runEntity.setFdStatus(1);
        runEntity.setFdEndTime(new Date());
        runEntity.setFdTime("" + (runEntity.getFdEndTime().getTime() - runEntity.getFdStartTime().getTime()));
        runEntity.setFdParam(getParam());//参数-需要抽象方法
        return runEntity;
    }

    private RunDetail getEndRunDetail(RunDetail runDetail) {
        runDetail.setFdStatus(1);
        runDetail.setFdEndTime(new Date());
        runDetail.setFdTime("" + (runDetail.getFdEndTime().getTime() - runDetail.getFdStartTime().getTime()));
        runDetail.setFdParam("");//参数-需要抽象方法
        return runDetail;
    }

    private String getBatId(DscSsc dscSsc){
        String uuid = UUID.randomUUID().toString().substring(0,6);
        String batId = "";
        if(dscSsc.getSscType() == 1){
            batId = "SYS_AUTO_" ;
        } else  {
            batId = "Manual_" ;

        }
        batId = batId + DateUtil.format(new Date(),DatePattern.PURE_DATETIME_MS_FORMAT);
        return batId + "_" + uuid;
    }

    private Run initRunEntity(DscSsc dscSsc) {
        Run runEntity = new Run();
        runEntity.setFdBatchId(getBatId(dscSsc));//uuid
        runEntity.setFdStartTime(new Date());//开始时间
        runEntity.setFdSscid(dscSsc.getSscid());//订阅id
        runEntity.setFdStatus(3);//状态
        runEntity.setFdType(dscSsc.getSscType());//自动还是手动
        runEntity.setFdProId(dscSsc.getFdProject());//应用
        runEntity.setCreateTime(new Date());//应用
        runEntity.setFdBuStartDate(runDate);//业务开始时间
        runEntity.setFdBuEndDate(endDate);//业务结束时间
        runEntity.setFdSscname(dscSsc.getSscName());//订阅名称
        return runEntity;
    }

    private String subErrorString(String message) {
        if (message.length() > 300) {
            return message.substring(0, 300);
        }
        return message;
    }

    protected SysTableinfo getSysTableinfo(String tableId) {
        QueryWrapper<SysTableinfo> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("fd_tableid", tableId);
        queryWrapper.notIn("fd_status", "2");
        return iSysTableinfoService.getOne(queryWrapper);
    }

    //验证
    abstract boolean verificationParam();

    //获取脱敏得订阅信息
    abstract List<DscSsc> getDscSsc();

    //获取脱敏运行组件
    abstract DscConversionService getComponent();

    //设置表格对应的字段信息
    abstract void setTableField(DscSscTableDetail dscSscTableDetail);

    //设置不同模式下的分区条件
    abstract void setWhere(DscSscTableDetail dscSscTableDetail);

    //返回参数
    abstract String getParam();


}
